﻿/*----------------------------------------------------------------------------------------
*
*  Copyright 2019, Gao Hai Hui, <fromzeropoint@126.com>.  All rights reserved.
*  https://gitee.com/helloworldghh/xoskit.git
*  Use of this source code is governed by a MIT license
*  that can be found in the License file.
*
----------------------------------------------------------------------------------------*/
#include "../import/head.h"
#include "../global/head.h"
#include "../macro/head.h"
#include "aio_impl.h"
#include "aio_tcp.h"

//  if(errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)
// 
/////////////////////////////////////////////////////////////////////////////////////////
// 
namespace xos_aio
{
    aio_tcp::aio_tcp( aio_impl * pAioImpl ) : m_pAioImpl( pAioImpl )
    {
    }

    aio_tcp::~aio_tcp()
    {
    }

    /////////////////////////////////////////////////////////////////////////////////////
    // 

    // 
    // m_bData[0] : true or false for result
    // m_bData[1] : true or false for server
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::user_init( xos_box::i_msg * lpMsg )
    {
        using namespace std;
        int nRet = 0;

        bool bIsServer = lpMsg->get_bool( 1, 0 );
        xos::i_socket * lpSocket = 0;
        int hSocket = -1;
        aio_key * lpKey = 0;

        if( 0 == nRet )
        {
            lpMsg->set_int( 0, i_aio::AIO_TCP_INIT_RET );
            lpMsg->set_bool( 0, false );
        }

        if( 0 == nRet )
        {
            nRet = g_pSysData->m_pContainer->xos()->create( xos::i_xos::XOS_OBJ_SOCKET, ( void ** )&lpSocket );
        }

        if( 0 == nRet )
        {
            const char * pszIp = lpMsg->get_str( 1, 0, 0, 0 );
            xos::xos_u16 usPort = ( xos::xos_u16 )lpMsg->get_int( 1, 0 );
            nRet = lpSocket->tcp_init( pszIp, usPort, true );
        }

        if( 0 == nRet )
        {
            xos::xos_u16 usPort = 0;
            char szIp[100] = { 0 };
            lpSocket->get_socket_addr( szIp, &usPort );
            lpMsg->set_str( 1, szIp );
            lpMsg->set_int( 1, usPort );
        }

        if( 0 == nRet )
        {
            hSocket = (int)lpSocket->get_socket();
            lpSocket->set_async( true );
        }

        if( ( 0 == nRet ) && bIsServer )
        {
            lpSocket->tcp_listen( 5 );
        }

        if( 0 == nRet )
        {
            lpKey = aio_key::get_item_from_pool( true );
            lpKey->init();
            lpKey->m_lpUserKey = lpMsg->get_void( 0, 0 );
            lpMsg->set_void( 1, lpKey );
            lpKey->m_enumHandleType = aio_key::H_TYPE_TCP_SERVER;
            lpKey->m_pSocket = lpSocket;
            lpKey->m_eState = aio_key::KEY_STATE_RUNNING;
        }

        if( 0 == nRet )
        {
            lpMsg->set_bool( 0, true );
            lpKey->m_nKqueueFlags = EVFILT_READ;
            struct kevent change[1];
            EV_SET( &change[0], hSocket, EVFILT_READ, EV_ADD, 0, 0, lpKey );
            kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
        }

        m_pAioImpl->notify( lpMsg );

        if( 0 == nRet )
        {
            lpMsg = 0;
            lpSocket = 0;
            lpKey = 0;
        }

        if( lpSocket )
        {
            lpSocket->release();
            lpSocket = 0;
        }

        if( lpKey )
        {
            lpKey->m_pSocket = 0;
            lpKey->release();
            lpKey = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // m_lpData[2] : ClientKey
    // 
    int aio_tcp::user_accept( xos_box::i_msg * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->get_void( 1, 0 );
        xos_box::i_msg * lpmsg = 0;

        lpMsg->set_int( 0, i_aio::AIO_TCP_ACCPET_RET );
        lpMsg->set_bool( 0, false );
        lpMsg->set_void( 2, 0 );

        if( lpKey->m_accept_list.size() > 0 )
        {
            aio_data * pData = lpKey->m_accept_list.front();
            lpKey->m_accept_list.pop_front();
            lpmsg = pData->m_pMsg;
            pData->m_pMsg = 0;
            pData->release();
            pData = 0;
        }

        if( lpmsg )
        {
            lpMsg->set_str( 1, lpmsg->get_str( 1, 0, 0, 0 ) );
            lpMsg->set_int( 1, lpmsg->get_int( 1, 0 ) );
            lpMsg->set_str( 2, lpmsg->get_str( 2, 0, 0, 0 ) );
            lpMsg->set_int( 2, lpmsg->get_int( 2, 0 ) );
            lpMsg->set_bool( 0, lpmsg->get_bool( 0, 0 ) );
            lpMsg->set_int( 0, lpmsg->get_int( 0, 0 ) );
            lpMsg->set_void( 2, lpmsg->get_void( 2, 0 ) );

            lpmsg->release();
            lpmsg = 0;
            m_pAioImpl->notify( lpMsg );
            lpMsg = 0;
        }
        else
        {
            aio_data * pData = aio_data::get_item_from_pool( true );
            pData->init();
            pData->m_pMsg = lpMsg;
            lpKey->m_pend_accept_list.push_back( pData );
            pData = 0;
            lpMsg = 0;
        }

        if( aio_key::KEY_STATE_RUNNING != lpKey->m_eState )
        {
            lpKey->finish_request( m_pAioImpl );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // m_lpData[2] : ClientKey
    // 
    int aio_tcp::tcp_accept( aio_key * lpKey, int nFilter )
    {
        using namespace std;
        int nRet = 0;

        xos::i_socket * lpSocket = lpKey->m_pSocket, * lpAcceptTcp = 0;
        aio_key * lpRetKey = 0;
        xos_box::i_msg * lpMsg = 0;
        bool bAddMsg = false;

        if( ( 0 == nRet ) && ( aio_key::KEY_STATE_RUNNING != lpKey->m_eState ) )
        {
            nRet = 1;
        }

        if( 0 == lpKey->m_pend_accept_list.size() )
        {
            lpMsg = g_pSysData->m_pContainer->msg();
            lpMsg->set_int( 0, i_aio::AIO_TCP_ACCPET_RET );
            lpMsg->set_bool( 0, false );
            lpMsg->set_void( 1, lpKey );
            lpMsg->set_void( 2, 0 );
            bAddMsg = true;
        }
        else
        {
            aio_data * p = lpKey->m_pend_accept_list.front();
            lpKey->m_pend_accept_list.pop_front();
            lpMsg = p->m_pMsg;
            p->m_pMsg = 0;
            p->release();
            p = 0;
        }

        if( 0 == nRet )
        {
            nRet = lpSocket->tcp_accept( &lpAcceptTcp );
            if( 0 == nRet )
            {
                lpAcceptTcp->set_async( true );
            }
        }

        if( 0 == nRet )
        {
            lpRetKey = aio_key::get_item_from_pool( true );
            lpRetKey->init();
        }

        if( 0 == nRet )
        {
            char szIp[100] = { 0 };
            xos::xos_u16 usPort = 0;
            lpAcceptTcp->get_socket_addr( szIp, &usPort );
            lpMsg->set_str( 1, szIp );
            lpMsg->set_int( 1, usPort );
        }

        if( 0 == nRet )
        {
            char szIp[100] = { 0 };
            xos::xos_u16 usPort = 0;
            lpAcceptTcp->get_peer_addr( szIp, &usPort );
            lpMsg->set_str( 2, szIp );
            lpMsg->set_int( 2, usPort );
        }

        if( 0 == nRet )
        {
            lpRetKey->m_nKqueueFlags = EVFILT_WRITE;
            lpRetKey->m_enumHandleType = aio_key::H_TYPE_TCP_NORM;
            lpRetKey->m_eState = aio_key::KEY_STATE_RUNNING;
            lpRetKey->m_pSocket = lpAcceptTcp;
            lpMsg->set_void( 2, lpRetKey );
        }

        if( 0 == nRet )
        {
            int hSocket = ( int )lpAcceptTcp->get_socket();
            struct kevent change[1];
            EV_SET( &change[0], hSocket, EVFILT_WRITE, EV_ADD, 0, 0, lpRetKey );
            kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
            EV_SET( &change[0], hSocket, EVFILT_READ, EV_ADD, 0, 0, lpRetKey );
            kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
        }

        if( 0 == nRet )
        {
            lpMsg->set_bool( 0, true );
            if( bAddMsg )
            {
                aio_data * p = aio_data::get_item_from_pool( true );
                p->init();
                p->m_pMsg = lpMsg;
                lpKey->m_accept_list.push_back( p );
                p = 0;
            }
            else
            {
                m_pAioImpl->notify( lpMsg );
            }
            lpAcceptTcp = 0;
            lpRetKey = 0;
            lpMsg = 0;
        }
        else if( lpMsg )
        {
            if( bAddMsg )
            {
                lpMsg->release();
            }
            else
            {
                m_pAioImpl->notify( lpMsg );
            }
            lpMsg = 0;
        }

        if( lpAcceptTcp )
        {
            lpAcceptTcp->release();
            lpAcceptTcp = 0;
        }

        if( lpRetKey )
        {
            lpRetKey->m_pSocket = 0;
            lpRetKey->release();
            lpRetKey = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // 
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::user_connect( xos_box::i_msg * lpMsg )
    {
        using namespace std;
        int nRet = 0;

        xos::i_socket * lpSocket = 0;
        aio_key * lpKey = 0;
        int hSocket = -1;

        lpMsg->set_int( 0, i_aio::AIO_TCP_CONNECT_RET );
        lpMsg->set_bool( 0, false );

        if( 0 == nRet )
        {
            nRet = g_pSysData->m_pContainer->xos()->create( xos::i_xos::XOS_OBJ_SOCKET, ( void ** )&lpSocket );
        }

        if( 0 == nRet )
        {
            const char * pszIp = lpMsg->get_str( 1, 0, 0, 0 );
            int nPort = lpMsg->get_int( 1, 0 );
            nRet = lpSocket->tcp_init( pszIp, nPort, true );
        }

        if( 0 == nRet )
        {
            hSocket = ( int )lpSocket->get_socket();
            lpSocket->set_async( true );
        }

        if( 0 == nRet )
        {
            lpKey = aio_key::get_item_from_pool( true );
            lpKey->init();
            lpKey->m_lpUserKey = lpMsg->get_void( 0, 0 );
            lpKey->m_enumHandleType = aio_key::H_TYPE_TCP_CONN;
            lpKey->m_eState = aio_key::KEY_STATE_RUNNING;
            lpKey->m_pSocket = lpSocket;
            lpKey->m_lpMsg = lpMsg;
            lpMsg->set_void( 1, lpKey );
        }

        if( 0 == nRet )
        {
            struct kevent change[1];
            EV_SET( &change[0], hSocket, EVFILT_WRITE, EV_ADD, 0, 0, lpKey );
            kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
            EV_SET( &change[0], hSocket, EVFILT_READ, EV_ADD, 0, 0, lpKey );
            kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
        }

        if( 0 == nRet )
        {
            unsigned long dwIp = 0;
            char szIp[100] = { 0 };
            lpSocket->get_ip_by_name( lpMsg->get_str( 2, 0, 0, 0 ), &dwIp, szIp );
            sockaddr_in addr = { 0 };
            addr.sin_family = AF_INET;
            addr.sin_addr.s_addr = dwIp;
            addr.sin_port = htons( lpMsg->get_int( 2, 0 ) );
            nRet = connect( hSocket, ( const sockaddr * )&addr, sizeof( sockaddr_in ) );
            if( ( 0 != nRet ) && ( EINPROGRESS == errno ) )
            {
                nRet = 0;
            }
        }

        if( 0 == nRet )
        {
            lpKey = 0;
            lpSocket = 0;
            lpMsg = 0;
        }

        if( lpMsg )
        {
            m_pAioImpl->notify( lpMsg );
            lpMsg = 0;
        }

        if( lpSocket )
        {
            lpSocket->release();
            lpSocket = 0;
        }

        if( lpKey )
        {
            lpKey->m_pSocket = 0;
            lpKey->release();
            lpKey = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : local port
    // m_szStr[1] : local ip
    // m_nData[2] : peer port
    // m_szStr[2] : peer ip
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_connect( aio_key * lpKey, int nFilter )
    {
        using namespace std;
        int nRet = 0;

        xos::i_socket * lpSocket = lpKey->m_pSocket;
        xos_box::i_msg * lpMsg = lpKey->m_lpMsg;

        if( ( 0 == nRet ) && ( aio_key::KEY_STATE_RUNNING != lpKey->m_eState ) )
        {
            nRet = 1;
        }

        if( 0 == nRet )
        {
            xos::xos_u16 usPeerPort = 0;
            char szPeerIp[100] = { 0 };
            lpSocket->get_peer_addr( szPeerIp, &usPeerPort );
            const char * ip = lpMsg->get_str( 2, 0, 0, 0 );
            int nPort = lpMsg->get_int( 2, 0 );
            if( ( ( int )usPeerPort != nPort ) || ( 0 != g_pSysData->m_pCrt->strcmp( szPeerIp, ip ) ) )
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            xos::xos_u16 usLocalPort = 0;
            char szLocalIp[100] = { 0 };
            nRet = lpSocket->get_socket_addr( szLocalIp, &usLocalPort );
            if( 0 == nRet )
            {
                lpMsg->set_str( 1, szLocalIp );
                lpMsg->set_int( 1, usLocalPort );
            }
        }

        if( 0 == nRet )
        {
            lpMsg->set_bool( 0, true );
            lpKey->m_enumHandleType = aio_key::H_TYPE_TCP_NORM;
            lpKey->m_bCanWrite = true;
            lpKey->m_lpMsg = 0;
            lpKey->m_eState = aio_key::KEY_STATE_RUNNING;
        }

        if( 0 != nRet )
        {
            lpMsg->set_void( 1, 0 );
            lpKey->release();
            lpKey = 0;
        }

        m_pAioImpl->notify( lpMsg );
        lpMsg = 0;

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : send bytes
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::user_send( xos_box::i_msg * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->get_void( 1, 0 );
        int ret = 0;

        lpMsg->set_int( 0, i_aio::AIO_TCP_SEND_RET );
        lpMsg->set_bool( 0, false );
        lpMsg->set_int( 1, 0 );

        {
            aio_data * pData = aio_data::get_item_from_pool( true );
            pData->init();
            pData->m_pMsg = lpMsg;
            lpKey->m_pend_send_list.push_back( pData );
            pData = 0;
            lpMsg = 0;
        }

        while( ( 0 == ret ) && ( aio_key::KEY_STATE_RUNNING == lpKey->m_eState ) )
        {
            ret = tcp_send( lpKey );
        }

        if( aio_key::KEY_STATE_RUNNING != lpKey->m_eState )
        {
            lpKey->finish_request( m_pAioImpl );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : send bytes
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_send( aio_key * lpKey )
    {
        using namespace std;
        int nRet = 0;

        xos::i_socket * lpSocket = lpKey->m_pSocket;
        xos_box::i_msg * lpMsg = 0;
        xos_box::i_buf * lpBuf = 0;
        int nResult = -1;

        if( ( 0 == nRet ) && ( aio_key::KEY_STATE_RUNNING != lpKey->m_eState ) )
        {
            nRet = 1;
        }

        if( ( 0 == nRet ) && !lpKey->m_bCanWrite )
        {
            nRet = 1;
        }

        if( 0 == nRet )
        {
			aio_data * p = 0;
			if( lpKey->m_pend_send_list.size() > 0 )
			{
				p = lpKey->m_pend_send_list.front();
                lpKey->m_pend_send_list.pop_front();
                lpMsg = p->m_pMsg;
                p->m_pMsg = 0;
                p->release();
                p = 0;
                lpBuf = lpMsg->get_buf( 0, 0 );
            }
            else
            {
                nRet = 1;
            }
        }

        if( 0 == nRet )
        {
            const char * pSendData = lpBuf->get_data( 0, 0, 0 );
            int len = lpBuf->get_len( 0 );
            int pos = lpBuf->get_pos( 0 );
            int ret = lpSocket->tcp_send( pSendData, len );
            if( ret > 0 )
            {
                int nSendLen = lpMsg->get_int( 1, 0 );
                nSendLen += ret;
                lpMsg->set_int( 1, nSendLen );
                pos += ret;
                len -= ret;
                lpBuf->set_len( len );
                lpBuf->set_pos( pos );
                if( 0 == len )
                {
                    lpMsg->set_bool( 0, true );
                    lpBuf->set_len( nSendLen );
                    lpBuf->set_pos( 0 );
                    nResult = 0;
                }
                else
                {
                    nResult = 1;
                }
            }
            else if( errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK )
            {
                lpKey->m_bCanWrite = false;
                nResult = 1;
                nRet = 1;
            }
            else
            {
                nResult = 2;
                nRet = 1;
            }
        }

        if( ( 0 == nResult ) || ( 2 == nResult ) )
        {
            m_pAioImpl->notify( lpMsg );
            lpMsg = 0;
        }

        if( lpMsg )
        {
            aio_data * p = aio_data::get_item_from_pool( true );
            p->init();
            p->m_pMsg = lpMsg;
            lpKey->m_pend_send_list.push_front( p );
            lpMsg = 0;
            p = 0;
        }

        if( 2 == nResult )
        {
            lpKey->finish_request( m_pAioImpl );
        }

        if( !lpKey->m_bCanWrite )
        {
            struct kevent change[1];
            EV_SET( &change[0], ( int )lpSocket->get_socket(), EVFILT_WRITE, EV_ENABLE, 0, 0, lpKey );
            kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
        }
        else if( 0 == lpKey->m_pend_send_list.size() )
        {
            struct kevent change[1];
            EV_SET( &change[0], ( int )lpSocket->get_socket(), EVFILT_WRITE, EV_DISABLE, 0, 0, lpKey );
            kevent( m_pAioImpl->m_hKqueue, change, 1, NULL, 0, NULL );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : recv len
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::user_recv( xos_box::i_msg * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = ( aio_key * )lpMsg->get_void( 1, 0 );
        xos_box::i_buf * lpBuf = lpMsg->get_buf( 0, 0 );
        xos_box::i_msg * lpmsg = 0;

        lpMsg->set_int( 0, i_aio::AIO_TCP_RECV_RET );
        lpMsg->set_bool( 0, false );
        lpMsg->set_int( 1, 0 );

        if( lpKey->m_recv_list.size() > 0 )
        {
            aio_data * p = lpKey->m_recv_list.front();
            lpKey->m_recv_list.pop_front();
            lpmsg = p->m_pMsg;
            p->m_pMsg = 0;
            p->release();
            p = 0;
        }

        if( lpmsg )
        {
            xos_box::i_buf * lpbuf = lpmsg->get_buf( 0, 0 );
            lpMsg->set_bool( 0, lpmsg->get_bool( 0, 0 ) );
            lpMsg->set_int( 0, lpmsg->get_int( 0, 0 ) );
            lpMsg->set_int( 1, lpmsg->get_int( 1, 0 ) );
            if( lpMsg->get_bool( 0, 0 ) )
            {
                lpBuf->set_data( lpbuf->get_data( 0, 0, 0 ), lpbuf->get_len( 0 ) );
            }
            m_pAioImpl->notify( lpMsg );

            lpMsg = 0;
            lpmsg->release();
            lpmsg = 0;
            lpbuf->release();
            lpbuf = 0;
        }
        else
        {
            aio_data * p = aio_data::get_item_from_pool( true );
            p->init();
            p->m_pMsg = lpMsg;
            lpKey->m_pend_recv_list.push_back( p );
            lpMsg = 0;
            p = 0;
        }

        if( aio_key::KEY_STATE_RUNNING != lpKey->m_eState )
        {
            lpKey->finish_request( m_pAioImpl );
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : recv len
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::tcp_recv( aio_key * lpKey )
    {
        using namespace std;
        int nRet = 0;

        xos::i_socket * lpSocket = lpKey->m_pSocket;
        xos_box::i_msg * lpMsg = 0;
        xos_box::i_buf * lpBuf = 0;
        bool bIsAddBuf = false;
        int nResult = -1;

        if( ( 0 == nRet ) && ( aio_key::KEY_STATE_RUNNING != lpKey->m_eState ) )
        {
            nRet = 1;
        }

        if( 0 == nRet )
        {
            if( 0 == lpKey->m_pend_recv_list.size() )
            {
                lpBuf = g_pSysData->m_pContainer->buf();
                lpMsg = g_pSysData->m_pContainer->msg();
                lpMsg->set_int( 0, i_aio::AIO_TCP_RECV_RET );
                lpMsg->set_buf( 0, lpBuf );
                lpMsg->set_void( 1, lpKey );
                lpMsg->set_bool( 0, false );
                lpMsg->set_int( 1, 0 );
                bIsAddBuf = true;
            }
            else
            {
                aio_data * p = lpKey->m_pend_recv_list.front();
                lpKey->m_pend_recv_list.pop_front();
                lpMsg = p->m_pMsg;
                p->m_pMsg = 0;
                p->release();
                p = 0;
                lpBuf = lpMsg->get_buf( 0, 0 );
            }
        }

        if( 0 == nRet )
        {
            char * pBuf = lpBuf->get_data( 0, 0, 0 );
            int nBufSize = lpBuf->get_buf_size( 0 );
            int len = lpSocket->tcp_recv( pBuf, nBufSize );
            if( len > 0 )
            {
                lpMsg->set_bool( 0, true );
                lpMsg->set_int( 1, len );
                lpBuf->set_len( len );
                if( nBufSize > len )
                {
                    nRet = 1;
                }
                nResult = 0;
            }
            else if( 0 == len )
            {
                nResult = 2;
                nRet = 1;
            }
            else if( errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK )
            {
                nResult = 1;
                nRet = 1;
            }
            else
            {
                lpMsg->set_int( 1, -1 );
                nResult = 2;
                nRet = 1;
            }
        }

        if( bIsAddBuf )
        {
            if( ( 0 == nResult ) || ( 2 == nResult ) )
            {
                aio_data * p = aio_data::get_item_from_pool( true );
                p->init();
                p->m_pMsg = lpMsg;
                lpKey->m_recv_list.push_back( p );
                lpMsg = 0;
            }
            else
            {
                if( lpMsg )
                {
                    lpMsg->release();
                    lpMsg = 0;
                }
                if( lpBuf )
                {
                    lpBuf->release();
                    lpBuf = 0;
                }
            }
        }
        else
        {
            if( ( 0 == nResult ) || ( 2 == nResult ) )
            {
                m_pAioImpl->notify( lpMsg );
                lpMsg = 0;
            }
            if( lpMsg )
            {
                aio_data * p = aio_data::get_item_from_pool( true );
                p->init();
                p->m_pMsg = lpMsg;
                lpKey->m_pend_recv_list.push_front( p );
                lpMsg = 0;
            }
        }

        if( 2 == nResult )
        {
            lpKey->finish_request( m_pAioImpl );
        }

        return nRet;
    }

    // 
    // m_nData[0] : cmd type
    // m_nData[1] : how
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::shut_down( xos_box::i_msg * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = (aio_key *)lpMsg->get_void( 1, 0 );
        xos::i_socket * lpSocket = lpKey->m_pSocket;
        int how = lpMsg->get_int( 1, 0 );

        lpMsg->set_int( 0, i_aio::AIO_TCP_SHUTDOWN_RET );
        lpMsg->set_bool( 0, true );

        lpSocket->shut_down( how );
        {
            aio_data * p = aio_data::get_item_from_pool( true );
            p->init();
            p->m_pMsg = lpMsg;
            m_pAioImpl->m_return_list.push_back( p );
            lpMsg = 0;
            p = 0;
        }

        return nRet;
    }

    // 
    // m_bData[0] : true or false for opt result
    // 
    // m_nData[0] : cmd type
    // m_nData[1] : linger.l_onoff
    // m_nData[2] : linger.l_linger
    // 
    // m_lpData[0] : UserKey
    // m_lpData[1] : AioKey
    // 
    int aio_tcp::close_net( xos_box::i_msg * lpMsg )
    {
        int nRet = 0;

        aio_key * lpKey = (aio_key *)lpMsg->get_void( 1, 0 );
        int onoff = lpMsg->get_int( 1, 0 );
        int linger = lpMsg->get_int( 2, 0 );
        xos::i_socket * lpSocket = lpKey->m_pSocket;

        lpMsg->set_int( 0, i_aio::AIO_TCP_CLOSE_RET );
        lpMsg->set_bool( 0, true );

        lpKey->m_eState = aio_key::KEY_STATE_CLOSING;
        lpSocket->close( onoff, linger );
        {
            aio_data * p = aio_data::get_item_from_pool( true );
            p->init();
            p->m_pMsg = lpMsg;
            m_pAioImpl->m_return_list.push_back( p );
            lpMsg = 0;
            p = 0;
        }

        return nRet;
    }

} // xos_aio
